#include "Avalon/IO/ChannelAdapterServerConnection.hpp"
#include "Avalon/IO/Channel.hpp"
#include "Avalon/IO/NotConnectedException.hpp"

using namespace Avalon;
using namespace Avalon::IO;
using namespace Avalon::Threading;
using namespace boost;
using namespace boost::posix_time;
using namespace boost::signals2;
using namespace std;

ChannelAdapterServerConnection::ChannelAdapterServerConnection(
    ServerConnection* connection, ChannelFactory* channelFactory)
    : m_connection(connection),
      m_channelFactory(channelFactory) {}

ChannelAdapterServerConnection::~ChannelAdapterServerConnection() {
  Close();
}

void ChannelAdapterServerConnection::Start() {
  boost::lock_guard<boost::recursive_mutex> lock(m_mutex);
  if(InternalIsConnected()) {
    return;
  }
  m_connection->Start();
  bool isOpen;
  connection closedConnection = m_connection->ConnectClosedSignal(bind(
    &ChannelAdapterServerConnection::OnClosed, this), &isOpen);
  m_connectionState.Initialize(isOpen, closedConnection);
}

Async<Channel*>::Ptr ChannelAdapterServerConnection::Accept() {
  boost::lock_guard<boost::recursive_mutex> lock(m_mutex);
  if(!InternalIsConnected()) {
    BOOST_THROW_EXCEPTION(NotConnectedException());
  }
  Async<Channel*>::Ptr acceptResult = m_connection->Accept();
  AsyncOwner<Channel*>::Ptr wrappedChannel = AsyncOwner<Channel*>::Create();
  BaseAsync::State acceptState;
  acceptResult->ConnectFinishedSignal(bind(
    &ChannelAdapterServerConnection::OnAccept, this, acceptResult,
    wrappedChannel), &acceptState);
  if(acceptState != BaseAsync::PENDING) {
    WrapResult(acceptResult, wrappedChannel);
  }
  return wrappedChannel;
}

bool ChannelAdapterServerConnection::IsConnected() {
  boost::lock_guard<boost::recursive_mutex> lock(m_mutex);
  return InternalIsConnected();
}

void ChannelAdapterServerConnection::Close() {
  boost::unique_lock<boost::recursive_mutex> lock(m_mutex);
  if(!InternalIsConnected()) {
    return;
  }
  lock.unlock();
  m_connection->Close();
  lock.lock();
  if(InternalIsConnected()) {
    m_closedCondition.wait(lock);
  }
}

connection ChannelAdapterServerConnection::ConnectClosedSignal(
    const ClosedSignal::slot_type& slot) {
  boost::lock_guard<boost::recursive_mutex> lock(m_mutex);
  return m_closedSignal.connect(slot);
}

connection ChannelAdapterServerConnection::ConnectClosedSignal(
    const ClosedSignal::slot_type& slot, bool* isConnected) {
  boost::lock_guard<boost::recursive_mutex> lock(m_mutex);
  *isConnected = InternalIsConnected();
  return m_closedSignal.connect(slot);
}

bool ChannelAdapterServerConnection::InternalIsConnected() {
  return m_connectionState.IsOpen();
}

void ChannelAdapterServerConnection::WrapResult(
    const Async<Channel*>::Ptr& result,
    const AsyncOwner<Channel*>::Ptr& wrappedChannel) {
  if(result->GetState() == BaseAsync::EXCEPTION) {
    wrappedChannel->CopyException(result);
  } else {
    Channel* channel = m_channelFactory->Create(result->Get());
    wrappedChannel->SetResult(channel);
  }
}

void ChannelAdapterServerConnection::OnAccept(
    const Async<Channel*>::Ptr& result,
    const AsyncOwner<Channel*>::Ptr& wrappedChannel) {
  boost::lock_guard<boost::recursive_mutex> lock(m_mutex);
  WrapResult(result, wrappedChannel);
}

void ChannelAdapterServerConnection::OnClosed() {
  boost::lock_guard<boost::recursive_mutex> lock(m_mutex);
  m_connectionState.SetClosed();
  m_closedCondition.notify_all();
  m_closedSignal();
}
